package com.atguigu.edu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.TableProcess;
import com.atguigu.edu.common.GmallConfig;
import com.atguigu.edu.util.DruidPhoenixDSUtil;
import com.atguigu.edu.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.SQLException;
import java.util.*;

public class MyBroadcastFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {
    private DruidDataSource druidDataSource;

    private MapStateDescriptor<String, TableProcess> tableConfigDescriptor;


    public MyBroadcastFunction(MapStateDescriptor<String, TableProcess> tableConfigDescriptor) {
        this.tableConfigDescriptor = tableConfigDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidPhoenixDSUtil.getDataSource();
    }

    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableConfigDescriptor);
        String table = value.getString("table");
        TableProcess tableProcess = broadcastState.get(table);

        if(tableProcess != null){
            JSONObject data = value.getJSONObject("data");
            String sinkTable = tableProcess.getSinkTable();
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumns(data,sinkColumns);

            data.put("sinkTable", sinkTable);
            out.collect(data);
        }
    }

    private void filterColumns(JSONObject data, String sinkColumns) {
        Set<Map.Entry<String, Object>> dataEntries = data.entrySet();
        List<String> cols = Arrays.asList(sinkColumns.split(","));
        // 删除sinkColumns中不存在的列
        Iterator<Map.Entry<String, Object>> iterator = dataEntries.iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, Object> next = iterator.next();
            if (!cols.contains(next.getKey())) {
                iterator.remove();
            }
        }

//        dataEntries.removeIf(next -> !cols.contains(next.getKey()));

    }

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        JSONObject jsonObject = JSON.parseObject(value);

        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableConfigDescriptor);

        String op = jsonObject.getString("op");

        if("d".equals(op)){
            TableProcess before = jsonObject.getObject("before", TableProcess.class);
            String sourceTable = before.getSourceTable();
            broadcastState.remove(sourceTable);
        }else{
            TableProcess after = jsonObject.getObject("after", TableProcess.class);

            String sourceTable = after.getSourceTable();
            String sinkTable = after.getSinkTable();
            String sinkColumns = after.getSinkColumns();
            String sinkPk = after.getSinkPk();
            String sinkExtend = after.getSinkExtend();

            broadcastState.put(sourceTable,after);
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
        }
    }

    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        // 为主键及扩展字段赋默认值
        if (sinkPk == null) {
            sinkPk = "id";
        }
        if (sinkExtend == null) {
            sinkExtend = "";
        }


        // 封装建表 SQL
        StringBuilder sql = new StringBuilder();
        sql.append("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append(" ( ");

        String[] cols = sinkColumns.split(",");

        // 遍历添加字段信息
        for (int i = 0; i < cols.length; i++) {
            // 判断当前是否为主键
            if (cols[i].equals(sinkPk)) {
                sql.append(cols[i])
                        .append(" varchar primary key ");
            } else {
                sql.append(cols[i])
                        .append(" varchar ");
            }
            // 最后一个不加逗号
            if (i < cols.length - 1){
                sql.append(",");
            }
        }
        sql.append(")").append(sinkExtend);

        String createStatement = sql.toString();
        System.out.println(createStatement);
        DruidPooledConnection conn = null;
        try {
            conn = druidDataSource.getConnection();
            PhoenixUtil.executeSQL(createStatement,conn);
        } catch (SQLException sqlException) {
            sqlException.printStackTrace();
            System.out.println("从 Druid 连接池获取连接对象异常");
        }

    }
}
